package com.jscloud.bigdata.flink.CEP;


import com.jscloud.bigdata.flink.tableapi.groupwindow.TempSensorData;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;

/**
 * * 描述
 *
 *   * 在我们操作某些银行APP的时候，经常会发现，如果上一个操作与下一个操作IP变换了（例如上一个操作使用的流量操作，下一个操作我连接上了wifi去操作，IP就会变换），那么APP就要求我们重新进行登录，避免由于IP变换产生的风险操作
 *
 * * 需求
 *
 *   - 用户上一个操作与下一个操作IP变换报警
 *
 * * 数据格式如下
 *
 *   * 从socket当中输入数据源
 *
 *   ~~~
 *   192.168.52.100,zhubajie,https://icbc.com.cn/login.html,2020-02-12 12:23:45
 *   192.168.54.172,tangseng,https://icbc.com.cn/login.html,2020-02-12 12:23:46
 *   192.168.145.77,sunwukong,https://icbc.com.cn/login.html,2020-02-12 12:23:47
 *   192.168.52.100,zhubajie,https://icbc.com.cn/transfer.html,2020-02-12 12:23:47
 *   192.168.54.172,tangseng,https://icbc.com.cn/transfer.html,2020-02-12 12:23:48
 *   192.168.145.77,sunwukong,https://icbc.com.cn/transfer.html,2020-02-12 12:23:49
 *   192.168.145.77,sunwukong,https://icbc.com.cn/save.html,2020-02-12 12:23:52
 *   192.168.52.100,zhubajie,https://icbc.com.cn/save.html,2020-02-12 12:23:53
 *   192.168.54.172,tangseng,https://icbc.com.cn/save.html,2020-02-12 12:23:54
 *   192.168.54.172,tangseng,https://icbc.com.cn/buy.html,2020-02-12 12:23:57
 *   192.168.145.77,sunwukong,https://icbc.com.cn/buy.html,2020-02-12 12:23:58
 *   192.168.52.100,zhubajie,https://icbc.com.cn/buy.html,2020-02-12 12:23:59
 *   ~~~
 *
 *   * 整理之后的格式如下：
 *   192.168.145.77,sunwukong,https://icbc.com.cn/login.html,2020-02-12 12:23:47
 *   192.168.145.77,sunwukong,https://icbc.com.cn/transfer.html,2020-02-12 12:23:49
 *   192.168.145.77,sunwukong,https://icbc.com.cn/save.html,2020-02-12 12:23:52
 *   192.168.145.77,sunwukong,https://icbc.com.cn/buy.html,2020-02-12 12:23:58
 *   192.168.89.189,sunwukong,https://icbc.com.cn/pay.html,2020-02-12 12:24:05
 *   192.168.89.189,sunwukong,https://icbc.com.cn/login.html,2020-02-12 12:24:07
 *
 *   192.168.52.100,zhubajie,https://icbc.com.cn/login.html,2020-02-12 12:23:45
 *   192.168.52.100,zhubajie,https://icbc.com.cn/transfer.html,2020-02-12 12:23:47
 *   192.168.52.100,zhubajie,https://icbc.com.cn/save.html,2020-02-12 12:23:53
 *   192.168.52.100,zhubajie,https://icbc.com.cn/buy.html,2020-02-12 12:23:59
 *
 *   192.168.54.172,tangseng,https://icbc.com.cn/login.html,2020-02-12 12:23:46
 *   192.168.54.172,tangseng,https://icbc.com.cn/transfer.html,2020-02-12 12:23:48
 *   192.168.54.172,tangseng,https://icbc.com.cn/save.html,2020-02-12 12:23:54
 *   ~~~
 */
public class CheckIpChangeByState {
        public static void main(String[] args) throws Exception {
                Logger.getLogger("org").setLevel(Level.ERROR);
                //1.获取stream的执行环境
                StreamExecutionEnvironment executionEnvironment= StreamExecutionEnvironment.getExecutionEnvironment();

                DataStreamSource<String> dataStreamSource = executionEnvironment.socketTextStream("bigdata01", 9999);

                dataStreamSource.map(new MapFunction<String, Tuple2<String, UserLoginBean>>() {
                        @Override
                        public Tuple2<String, UserLoginBean> map(String s) throws Exception {
                                String[] split = s.split(",");
                                UserLoginBean userLoginBean = new UserLoginBean(split[0], split[1], split[2], split[3]);

                                Tuple2<String, UserLoginBean> stringUserLoginBeanTuple2 = new Tuple2<>();
                                stringUserLoginBeanTuple2.setFields(split[1], userLoginBean );
                                return stringUserLoginBeanTuple2;
                        }
                }).keyBy(new KeySelector<Tuple2<String, UserLoginBean>, String>() {
                        @Override
                        public String getKey(Tuple2<String, UserLoginBean> stringUserLoginBeanTuple2) throws Exception {
                                return stringUserLoginBeanTuple2.f0;
                        }
                }).process(new LoginCheckProcessFunction()).print();

                executionEnvironment.execute();
        }


        public static class LoginCheckProcessFunction extends KeyedProcessFunction<String, Tuple2<String, UserLoginBean>, Tuple2<String, UserLoginBean>> {
                private ListState<UserLoginBean> listState = null;

                @Override
                public void open(Configuration parameters) throws Exception {
                       listState = getRuntimeContext().getListState(new ListStateDescriptor<UserLoginBean>("changeIp", UserLoginBean.class));
                }

                @Override
                public void processElement(Tuple2<String, UserLoginBean> value,
                                           KeyedProcessFunction<String, Tuple2<String, UserLoginBean>, Tuple2<String, UserLoginBean>>.Context context, Collector<Tuple2<String, UserLoginBean>> collector) throws Exception {
                        ArrayList<UserLoginBean> sortList = new ArrayList<>();
                        ArrayList<UserLoginBean> userLoginBeans = new ArrayList<>();

                        UserLoginBean f1 = value.f1;
                        listState.add(f1);

                        //按照数据的时间排序
                        Iterator<UserLoginBean> iterator = listState.get().iterator();
                        while (iterator.hasNext()) {
                                UserLoginBean next = iterator.next();
                                sortList.add(next);
                        }
                        //进行排序
                        Collections.sort(sortList, new Comparator<UserLoginBean>() {
                                @Override
                                public int compare(UserLoginBean o1, UserLoginBean o2) {
                                        return o1.getTime().compareTo(o2.getTime());
                                }
                        });

                        if(sortList.size() == 2) {
                                UserLoginBean first = sortList.get(0);
                                UserLoginBean second = sortList.get(1);

                                if(first.getIp().equals(second.getIp())) {
                                        System.out.println("你的 IP 发生了变化，小心诈骗，重新登录");
                                }

                                sortList.removeAll(Collections.EMPTY_LIST);
                                userLoginBeans.add(second);
                                listState.update(userLoginBeans);
                        }
                        collector.collect(value);
                }
        }
}
